package fliter;

import data.DataPoint;
import extractor.SingleExtractor;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.SerializableFunction;
import org.apache.kafka.common.protocol.types.Field;
import org.joda.time.Instant;
import org.joda.time.DateTime;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

/**
 * Created by emily on 17-12-12.
 */
public class SensorIdFliter extends DoFn<DataPoint, DataPoint> {

    private static final Logger LOG = LoggerFactory.getLogger(SensorIdFliter.class);
    List<String> sensors;

    public SensorIdFliter(String input){
        sensors = Arrays.asList(input.split(","));
    }

    @ProcessElement
    public void processElement(ProcessContext c){
        if(sensors.contains(c.element().sensorId)){
            DateTime t = new DateTime(c.element().timestamp);
            c.output(c.element());
        }else{
            LOG.info("not contain");
        }
    }
}
